package drds.binlog.deployer;

import com.google.common.base.Function;
import com.google.common.collect.ComputingMap;
import com.google.common.collect.MapMaker;
import drds.binlog.common.utils.AddressUtils;
import drds.binlog.common.zookeeper.ZooKeeperClient;
import drds.binlog.common.zookeeper.Zookeepers;
import drds.binlog.common.zookeeper.running.Monitor;
import drds.binlog.common.zookeeper.running.ServerRunningData;
import drds.binlog.common.zookeeper.running.ServerRunningListener;
import drds.binlog.common.zookeeper.running.ServerRunningMonitors;
import drds.binlog.deployer.monitor.InstanceAction;
import drds.binlog.deployer.monitor.InstanceConfigMonitor;
import drds.binlog.deployer.monitor.ManagerInstanceConfigMonitor;
import drds.binlog.instance.core.BinlogTask;
import drds.binlog.instance.core.BinlogTaskFactory;
import drds.binlog.instance.manager.ConfigClient;
import drds.binlog.instance.manager.ManagerBinlogTaskFactory;
import drds.binlog.server.embedded.EmbeddedServer;
import drds.binlog.server.exception.BinlogServerException;
import drds.binlog.server.netty.NettyServer;
import org.I0Itec.zkclient.IZkStateListener;
import org.I0Itec.zkclient.exception.ZkNoNodeException;
import org.I0Itec.zkclient.exception.ZkNodeExistsException;
import org.apache.commons.lang.BooleanUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;

import java.util.Map;
import java.util.Properties;


public class Controller
{

    private static final Logger logger = LoggerFactory.getLogger(Controller.class);
    private Long cid;
    private String ip;
    private int port;
    // 默认使用spring的方式载入
    private Map<String, InstanceConfig> instanceConfigs;
    private InstanceConfig globalInstanceConfig;
    private Map<String, ConfigClient> managerClients;
    // 监听instance config的变化
    private boolean autoScan = true;
    private InstanceAction defaultAction;
    private InstanceConfigMonitor instanceConfigMonitors;
    private EmbeddedServer embeddedServer;
    private NettyServer nettyServer;

    private BinlogTaskFactory binlogTaskFactory;
    private ZooKeeperClient zooKeeperClient;

    public Controller()
    {
        this(System.getProperties());
    }

    public Controller(final Properties properties)
    {
        managerClients = ComputingMap.makeComputingMap(new Function<String, ConfigClient>()
        {

            public ConfigClient apply(String managerAddress)
            {
                return getManagerClient(managerAddress);
            }
        });

        // 初始化全局参数设置
        globalInstanceConfig = initGlobalConfig(properties);
        instanceConfigs = new MapMaker().makeMap();
        // 初始化instance config
        initInstanceConfig(properties);

        // init socketChannel
        String socketChannel = getProperty(properties, Constants.SOCKETCHANNEL);
        if (StringUtils.isNotEmpty(socketChannel))
        {
            System.setProperty(Constants.SOCKETCHANNEL, socketChannel);
        }

        // 准备canal server
        cid = Long.valueOf(getProperty(properties, Constants.ID));
        ip = getProperty(properties, Constants.IP);
        port = Integer.valueOf(getProperty(properties, Constants.PORT));
        embeddedServer = EmbeddedServer.instance();
        embeddedServer.setBinlogTaskFactory(binlogTaskFactory);// 设置自定义的instanceGenerator
        try
        {
            int metricsPort = Integer.valueOf(getProperty(properties, Constants.METRICS_PULL_PORT));
            embeddedServer.setMetricsPort(metricsPort);
        } catch (NumberFormatException e)
        {
            logger.info("No valid metrics server port found, use default 11112.");
            embeddedServer.setMetricsPort(11112);
        }

        String canalWithoutNetty = getProperty(properties, Constants.WITHOUT_NETTY);
        if (canalWithoutNetty == null || "false".equals(canalWithoutNetty))
        {
            nettyServer = NettyServer.instance();
            nettyServer.setIp(ip);
            nettyServer.setPort(port);
        }

        // 处理下ip为空，默认使用hostIp暴露到zk中
        if (StringUtils.isEmpty(ip))
        {
            ip = AddressUtils.getHostIp();
        }
        final String zkServers = getProperty(properties, Constants.ZKSERVERS);
        if (StringUtils.isNotEmpty(zkServers))
        {
            zooKeeperClient = ZooKeeperClient.getZkClient(zkServers);
            // 初始化系统目录
            zooKeeperClient.createPersistent(Zookeepers.DESTINATION_ROOT_NODE, true);
            zooKeeperClient.createPersistent(Zookeepers.CANAL_CLUSTER_ROOT_NODE, true);
        }

        final ServerRunningData serverData = new ServerRunningData(cid, ip + ":" + port);
        ServerRunningMonitors.setServerData(serverData);
        ServerRunningMonitors.setRunningMonitors(ComputingMap.makeComputingMap(new Function<String, Monitor>()
        {

            public Monitor apply(final String destination)
            {
                Monitor runningMonitor = new Monitor(serverData);
                runningMonitor.setDestination(destination);
                runningMonitor.setListener(new ServerRunningListener()
                {

                    public void processActiveEnter()
                    {
                        try
                        {
                            MDC.put(Constants.mdc_destination, String.valueOf(destination));
                            embeddedServer.start(destination);
                        } finally
                        {
                            MDC.remove(Constants.mdc_destination);
                        }
                    }

                    public void processActiveExit()
                    {
                        try
                        {
                            MDC.put(Constants.mdc_destination, String.valueOf(destination));
                            embeddedServer.stop(destination);
                        } finally
                        {
                            MDC.remove(Constants.mdc_destination);
                        }
                    }

                    public void processStart()
                    {
                        try
                        {
                            if (zooKeeperClient != null)
                            {
                                final String path = Zookeepers.getDestinationClusterNode(destination, ip + ":"
                                        + port);
                                initCid(path);
                                zooKeeperClient.subscribeStateChanges(new IZkStateListener()
                                {

                                    public void handleStateChanged(KeeperState state) throws Exception
                                    {

                                    }

                                    public void handleNewSession() throws Exception
                                    {
                                        initCid(path);
                                    }

                                    @Override
                                    public void handleSessionEstablishmentError(Throwable error) throws Exception
                                    {
                                        logger.error("failed to connect to zookeeper", error);
                                    }
                                });
                            }
                        } finally
                        {
                            MDC.remove(Constants.mdc_destination);
                        }
                    }

                    public void processStop()
                    {
                        try
                        {
                            MDC.put(Constants.mdc_destination, String.valueOf(destination));
                            if (zooKeeperClient != null)
                            {
                                final String path = Zookeepers.getDestinationClusterNode(destination, ip + ":"
                                        + port);
                                releaseCid(path);
                            }
                        } finally
                        {
                            MDC.remove(Constants.mdc_destination);
                        }
                    }

                });
                if (zooKeeperClient != null)
                {
                    runningMonitor.setZooKeeperClient(zooKeeperClient);
                }
                // 触发创建一下cid节点
                runningMonitor.init();
                return runningMonitor;
            }
        }));

        // 初始化monitor机制
        autoScan = BooleanUtils.toBoolean(getProperty(properties, Constants.AUTO_SCAN));
        if (autoScan)
        {
            defaultAction = new InstanceAction()
            {

                public void start(String destination)
                {
                    InstanceConfig config = instanceConfigs.get(destination);
                    if (config == null)
                    {
                        // 重新读取一下instance config
                        config = parseInstanceConfig(properties, destination);
                        instanceConfigs.put(destination, config);
                    }

                    if (!embeddedServer.isStart(destination))
                    {
                        // HA机制启动
                        Monitor runningMonitor = ServerRunningMonitors.getRunningMonitor(destination);
                        if (!config.getLazy() && !runningMonitor.isRunning())
                        {
                            runningMonitor.start();
                        }
                    }
                }

                public void stop(String destination)
                {
                    // 此处的stop，代表强制退出，非HA机制，所以需要退出HA的monitor和配置信息
                    InstanceConfig config = instanceConfigs.remove(destination);
                    if (config != null)
                    {
                        embeddedServer.stop(destination);
                        Monitor runningMonitor = ServerRunningMonitors.getRunningMonitor(destination);
                        if (runningMonitor.isRunning())
                        {
                            runningMonitor.stop();
                        }
                    }
                }

                public void reload(String destination)
                {
                    // 目前任何配置变化，直接重启，简单处理
                    stop(destination);
                    start(destination);
                }
            };

            instanceConfigMonitors = new ManagerInstanceConfigMonitor();
        }
    }

    private InstanceConfig initGlobalConfig(Properties properties)
    {
        InstanceConfig globalConfig = new InstanceConfig();

        String lazyStr = getProperty(properties, Constants.getInstancLazyKey(Constants.GLOBAL_NAME));
        if (StringUtils.isNotEmpty(lazyStr))
        {
            globalConfig.setLazy(Boolean.valueOf(lazyStr));
        }

        String managerAddress = getProperty(properties,
                Constants.getInstanceManagerAddressKey(Constants.GLOBAL_NAME));
        if (StringUtils.isNotEmpty(managerAddress))
        {
            globalConfig.setManagerAddress(managerAddress);
        }


        binlogTaskFactory = new BinlogTaskFactory()
        {

            public BinlogTask generate(String destination)
            {
                InstanceConfig config = instanceConfigs.get(destination);
                if (config == null)
                {
                    throw new BinlogServerException("can't find destination:{}");
                }

                ManagerBinlogTaskFactory instanceGenerator = new ManagerBinlogTaskFactory();
                instanceGenerator.setConfigClient(managerClients.get(config.getManagerAddress()));
                return instanceGenerator.generate(destination);

            }

        };

        return globalConfig;
    }

    private ConfigClient getManagerClient(String managerAddress)
    {
        return new ConfigClient();
    }


    private void initInstanceConfig(Properties properties)
    {
        String destinationStr = getProperty(properties, Constants.DESTINATIONS);
        String[] destinations = StringUtils.split(destinationStr, Constants.DESTINATION_SPLIT);

        for (String destination : destinations)
        {
            InstanceConfig config = parseInstanceConfig(properties, destination);
            InstanceConfig oldConfig = instanceConfigs.put(destination, config);

            if (oldConfig != null)
            {
                logger.warn("destination:{} old config:{} has replace by new config:{}", new Object[]{destination,
                        oldConfig, config});
            }
        }
    }

    private InstanceConfig parseInstanceConfig(Properties properties, String destination)
    {
        InstanceConfig config = new InstanceConfig(globalInstanceConfig);


        String lazyStr = getProperty(properties, Constants.getInstancLazyKey(destination));
        if (!StringUtils.isEmpty(lazyStr))
        {
            config.setLazy(Boolean.valueOf(lazyStr));
        }

        String managerAddress = getProperty(properties, Constants.getInstanceManagerAddressKey(destination));
        if (StringUtils.isNotEmpty(managerAddress))
        {
            config.setManagerAddress(managerAddress);
        }
        return config;
    }

    public String getProperty(Properties properties, String key)
    {
        key = StringUtils.trim(key);
        String value = System.getProperty(key);

        if (value == null)
        {
            value = System.getenv(key);
        }

        if (value == null)
        {
            value = properties.getProperty(key);
        }

        return StringUtils.trim(value);
    }

    public void start() throws Throwable
    {
        logger.info("## start the canal server[{}:{}]", ip, port);
        // 创建整个canal的工作节点
        final String path = Zookeepers.getCanalClusterNode(ip + ":" + port);
        initCid(path);
        if (zooKeeperClient != null)
        {
            this.zooKeeperClient.subscribeStateChanges(new IZkStateListener()
            {

                public void handleStateChanged(KeeperState state) throws Exception
                {

                }

                public void handleNewSession() throws Exception
                {
                    initCid(path);
                }

                @Override
                public void handleSessionEstablishmentError(Throwable error) throws Exception
                {
                    logger.error("failed to connect to zookeeper", error);
                }
            });
        }
        // 优先启动embeded服务
        embeddedServer.start();
        // 尝试启动一下非lazy状态的通道
        for (Map.Entry<String, InstanceConfig> entry : instanceConfigs.entrySet())
        {
            final String destination = entry.getKey();
            InstanceConfig config = entry.getValue();
            // 创建destination的工作节点
            if (!embeddedServer.isStart(destination))
            {
                // HA机制启动
                Monitor runningMonitor = ServerRunningMonitors.getRunningMonitor(destination);
                if (!config.getLazy() && !runningMonitor.isRunning())
                {
                    runningMonitor.start();
                }
            }

            if (autoScan)
            {
                instanceConfigMonitors.register(destination, defaultAction);
            }
        }

        if (autoScan)
        {
            instanceConfigMonitors.start();
            if (!instanceConfigMonitors.isRunning())
            {
                instanceConfigMonitors.start();
            }
        }

        // 启动网络接口
        if (nettyServer != null)
        {
            nettyServer.start();
        }
    }

    public void stop() throws Throwable
    {
        if (nettyServer != null)
        {
            nettyServer.stop();
        }

        if (autoScan)
        {
            if (instanceConfigMonitors.isRunning())
            {
                instanceConfigMonitors.stop();
            }
        }

        for (Monitor runningMonitor : ServerRunningMonitors.getRunningMonitors().values())
        {
            if (runningMonitor.isRunning())
            {
                runningMonitor.stop();
            }
        }

        // 释放canal的工作节点
        releaseCid(Zookeepers.getCanalClusterNode(ip + ":" + port));
        logger.info("## stop the canal server[{}:{}]", ip, port);

        if (zooKeeperClient != null)
        {
            zooKeeperClient.close();
        }
    }

    private void initCid(String path)
    {
        // logger.info("## init the binlogTaskId = {}", cid);
        // 初始化系统目录
        if (zooKeeperClient != null)
        {
            try
            {
                zooKeeperClient.createEphemeral(path);
            } catch (ZkNoNodeException e)
            {
                // 如果父目录不存在，则创建
                String parentDir = path.substring(0, path.lastIndexOf('/'));
                zooKeeperClient.createPersistent(parentDir, true);
                zooKeeperClient.createEphemeral(path);
            } catch (ZkNodeExistsException e)
            {
                // ignore
                // 因为第一次启动时创建了cid,但在stop/start的时可能会关闭和新建,允许出现NodeExists问题s
            }

        }
    }

    private void releaseCid(String path)
    {
        // logger.info("## release the binlogTaskId = {}", cid);
        // 初始化系统目录
        if (zooKeeperClient != null)
        {
            zooKeeperClient.delete(path);
        }
    }

}
